package com.yeeiee;

import com.yeeiee.config.JdbcCatalogConfig;
import com.yeeiee.config.MysqlSourceConfig;
import com.yeeiee.utils.FactoryUtils;
import lombok.val;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;

public class CatalogTest2 {
    public static void main(String[] args) throws TableNotExistException, TableAlreadyExistException, DatabaseNotExistException {
        val env = StreamExecutionEnvironment.getExecutionEnvironment();
        val tabEnv = StreamTableEnvironment.create(env);

        // mysql catalog
        val catalogConfig = new Configuration();
        catalogConfig.set(JdbcCatalogConfig.url, "jdbc:mysql://localhost:3306/chen_test");
        catalogConfig.set(JdbcCatalogConfig.username, "root");
        catalogConfig.set(JdbcCatalogConfig.password, "123");
        val catalog = FactoryUtils.getCatalog("mysql", catalogConfig);
        tabEnv.registerCatalog("mysql", catalog);


        val sourceConfig = new Configuration();
        sourceConfig.set(MysqlSourceConfig.catalog, "mysql");
        sourceConfig.set(MysqlSourceConfig.tablePath, "chen_test.t_user");
        val mysqlSource = FactoryUtils.getSource("mysqlSource", sourceConfig);

        mysqlSource.setConnector(catalog);

        tabEnv.executeSql("select * from " + mysqlSource.getTableFullPath()).print();

    }
}
